/*
 * service_site_manager.cpp
 *
 *  Created on: 2024-06-14
 *      Author: van
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <net/if.h>
#include <arpa/inet.h>
#include <sys/types.h>
#include <sys/ioctl.h>
#include <sys/socket.h>
#include <error.h>
#include <stdarg.h>
#include <unistd.h>
#include <mutex>
#include <condition_variable>
#include "http/httplib.h"
#include "nlohmann/json.hpp"
#include "service_site_manager.h"
#include "lock_utils.h"

//放在代码而不是头文件中, 确保使用so库时版本号一定是动态获取的
#define SERVICE_SITE_LIB_VER "2024-10-10"

const string OK_RESPONSE_JSON = "{\"code\": 0, \"error\": \"ok\"}";

const string ERROR_RESPONSE_JSON_FORMAT = "{\"code\": -2, \"error\": \"lib json format error\"}";
const string ERROR_RESPONSE_NOT_SERVICE_OR_MESSAGE = "{\"code\": -3, \"error\": \"not service or message\"}";
const string ERROR_RESPONSE_RESPONSE_IS_NULL = "{\"code\": -4, \"error\": \"response is null\"}";
const string ERROR_RESPONSE_REQUEST_HANDLER_ERROR = "{\"code\": -5, \"error\": \"request handler error\"}";
const string ERROR_RESPONSE_NO_REQUEST_HANDLER_MATCH = "{\"code\": -6, \"error\": \"no request handler match\"}";
const string ERROR_RESPONSE_REQUEST_ILLEGAL = "{\"code\": -7, \"error\": \"request illegal\"}";
const string ERROR_RESPONSE_NO_MESSAGE_HANDLER_MATCH = "{\"code\": -8, \"error\": \"no message handler match\"}";
const string ERROR_RESPONSE_HANDLER_ERROR = "{\"code\": -9, \"error\": \"handler error\"}";

using namespace std;
using namespace servicesite;
using namespace httplib;

using json = nlohmann::json;

const string ServiceSiteManager::RET_OK = "ok";

const string ServiceSiteManager::QUERY_SITE_ID = "site_query";
const string ServiceSiteManager::QUERY_SITE_IP = "127.0.0.1";

const string ServiceSiteManager::QUERY_SITE_MESSAGE_ID_REGISTER_AGAIN = "register2QuerySiteAgain";

const string ServiceSiteManager::SERVICE_ID_GET_SERVICE_LIST = "get_service_list";
const string ServiceSiteManager::SERVICE_ID_GET_MESSAGE_LIST = "get_message_list";
const string ServiceSiteManager::SERVICE_ID_DEBUG = "debug";
const string ServiceSiteManager::SERVICE_ID_SUBSCRIBE_MESSAGE = "subscribe_message";
const string ServiceSiteManager::SERVICE_ID_UNSUBSCRIBE_MESSAGE = "unsubscribe_message";

const string ServiceSiteManager::ACCESS_SERVICE_ID_GET_MESSAGE_LIST = "get_message_list";
const string ServiceSiteManager::ACCESS_SERVICE_ID_SUBSCRIBE_MESSAGE = "subscribe_message";
const string ServiceSiteManager::ACCESS_SERVICE_ID_UNSUBSCRIBE_MESSAGE = "unsubscribe_message";
const string ServiceSiteManager::ACCESS_SERVICE_ID_SITE_PING = "site_ping";
const string ServiceSiteManager::ACCESS_SERVICE_ID_SITE_REGISTER = "site_register";

const string ServiceSiteManager::MESSAGE_SUBSCRIBER_CONFIG_FILE = "_message_subscriber.json";
const string ServiceSiteManager::MESSAGE_SUBSCRIBER_CONFIG_FILE_PATH = "/data/changhong/edge_midware/";

ServiceSiteManager ServiceSiteManager::instance;

int ServiceSiteManager::registerServiceRequestHandler(string handler_service_id, ServiceRequestHandler handler) {
    if (handler_service_id.empty())
    {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "registerServiceRequestHandler, handler_service_id is empty");
        return -1;
    }
    pthread_rwlock_guard lock(init_mutex, true);
    for (auto &x : serviceRequestHandlers) {
        if (x.first != handler_service_id)
            continue;

        x.second = handler;
        handle_log(LOG_LEVEL_WARN, __FILE__, __LINE__, "registerServiceRequestHandler, dup service handler registed, only the last one will be used: handler_service_id=" + handler_service_id);
        return RET_CODE_OK;
    }
    serviceRequestHandlers.push_back(std::make_pair(handler_service_id, handler));

    return RET_CODE_OK;
}

int ServiceSiteManager::registerMessageId(string messageId) {
    return registerMessageId(messageId, messageId, messageId);
}

int ServiceSiteManager::registerMessageId(string messageId, string name, string summary) {
    if (messageId.empty())
    {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "registerMessageHandler, messageId is empty");
        return -1;
    }
    if ("*" == messageId)
    {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "registerMessageHandler, messageId can not be `*`");
        return -1;
    }
    MessageT message_id;

    message_id.messageId = messageId;
    message_id.name = name;
    message_id.summary = summary;

    pthread_rwlock_guard lock(init_mutex, true);
    for (auto &x : messageIds) {
        if (x.messageId != messageId)
            continue;

        x.name = name;
        x.summary = summary;
        handle_log(LOG_LEVEL_WARN, __FILE__, __LINE__, "registerMessageId, dup message id registed, only the last one will be used: messageId=" + messageId);
        return RET_CODE_OK;
    }
    messageIds.push_back(message_id);

    return RET_CODE_OK;
}

int ServiceSiteManager::serviceRequestHandlerGetServiceList(const Request& request, Response& response) {
    ServiceSiteManager* that = ServiceSiteManager::getInstance();
    ServiceGetServiceListResponse response_obj;

    {
        pthread_rwlock_guard lock(that->init_mutex, false);
        for (const auto& item : that->serviceRequestHandlers) {
            response_obj.service_list.push_back(item.first);
        }
    }

    ServiceSiteLibServiceResponse service_site_response;

    service_site_response.code = 0;
    service_site_response.error = "ok";
    service_site_response.response = response_obj.to_json();

    response.set_content(service_site_response.to_json().dump(), "text/json");

    return RET_CODE_OK;
}

int ServiceSiteManager::serviceRequestHandlerGetMessageList(const Request& request, Response& response) {
    ServiceSiteManager* that = ServiceSiteManager::getInstance();
    ServiceGetMessageListResponse response_obj;

    {
        pthread_rwlock_guard lock(that->init_mutex, false);
        for (const auto& item : that->messageIds) {
            ServiceGetMessageListResponseMessageListItem msg_item_obj;

            msg_item_obj.message_id = item.messageId;
            msg_item_obj.name = item.name;
            msg_item_obj.summary = item.summary;

            response_obj.message_list.push_back(msg_item_obj);
        }
    }

    ServiceSiteLibServiceResponse service_site_response;

    service_site_response.code = 0;
    service_site_response.error = "ok";
    service_site_response.response = response_obj.to_json();

    response.set_content(service_site_response.to_json().dump(), "text/json");

    return RET_CODE_OK;
}

int ServiceSiteManager::serviceRequestHandlerSubscribeMessage(const Request& request, Response& response) {
    ServiceSubscribeMessageResponse response_obj;
    
    ServiceSiteLibServiceResponse service_site_response;

    service_site_response.code = 0;
    service_site_response.error = "ok";
    service_site_response.response = response_obj.to_json();

    // string ip = request.remote_addr;

    json request_json = json::parse(request.body);

    bool need_save = false;

    if (request_json["request"].is_null()) {
        service_site_response.code = -101;
        service_site_response.error = "request is null";
        response.set_content(service_site_response.to_json().dump(), "text/json");
        return RET_CODE_OK;
    }

    ServiceSubscribeMessageRequest request_obj;

    int ret = request_obj.from_json(request_json["request"]);
    if (ret != 0) {
        service_site_response.code = -102;
        service_site_response.error = "request json format error";
        response.set_content(service_site_response.to_json().dump(), "text/json");
        return RET_CODE_OK;
    }

    for (auto& message_id : request_obj.message_list) {
        need_save |= ServiceSiteManager::getInstance()->handleSubscribeMessage(message_id, request_obj.mac, request_obj.site_id, request_obj.port, request_obj.ip);
    }

    if (need_save) {
        ServiceSiteManager::getInstance()->saveMessageSubscriber();
    }
    
    response.set_content(OK_RESPONSE_JSON, "text/json");
    return RET_CODE_OK;
}

int ServiceSiteManager::serviceRequestHandlerUnsubscribeMessage(const Request& request, Response& response) {
    ServiceUnsubscribeMessageResponse response_obj;
    
    ServiceSiteLibServiceResponse service_site_response;

    service_site_response.code = 0;
    service_site_response.error = "ok";
    service_site_response.response = response_obj.to_json();

    // string ip = request.remote_addr;

    json request_json = json::parse(request.body);


    if (request_json["request"].is_null()) {
        service_site_response.code = -101;
        service_site_response.error = "request is null";
        response.set_content(service_site_response.to_json().dump(), "text/json");
        return RET_CODE_OK;
    }

    ServiceUnsubscribeMessageRequest request_obj;
    int ret = request_obj.from_json(request_json["request"]);
    if (ret != 0) {
        service_site_response.code = -102;
        service_site_response.error = "request json format error";
        response.set_content(service_site_response.to_json().dump(), "text/json");
        return RET_CODE_OK;
    }

    bool need_save = false;
    ServiceSiteManager* that = ServiceSiteManager::getInstance();
    {
        pthread_rwlock_guard lock(that->messageSubscriberList_mutex, true);
        MessageSubscriberSiteHandle* temp_messageSubscriberSiteHandle = NULL;
        for (auto& item : that->messageSubscriberSiteHandlePList) {
            if (item->mac == request_obj.mac) {
                if (item->site_id == request_obj.site_id) {
                    temp_messageSubscriberSiteHandle = item;
                    break;
                }
            }
        }
        // 此站点没有订阅过， 忽略
        if (temp_messageSubscriberSiteHandle != NULL) {
            for (auto& json_message_id : request_obj.message_list) {
                MessageSubscriber* temp_messageSubscriber = NULL;
                for (auto& item : that->messageSubscriberList) {
                    if (json_message_id == item.messageId) {
                        temp_messageSubscriber = &item;
                        break;
                    }
                }
                if (temp_messageSubscriber == NULL) {
                    // 此消息没有订阅过， 忽略
                    continue;
                }
                else {
                    temp_messageSubscriber->delSiteMessageSubscriberSiteHandleP(temp_messageSubscriberSiteHandle);

                    need_save = true;
                }
            }
        }
    }

    response.set_content(OK_RESPONSE_JSON, "text/json");

    if (need_save) {
        that->saveMessageSubscriber();
    }

    return RET_CODE_OK;
}

//TODO: 考虑加个配置, 每隔固定时间在日志中打印下统计数据
int ServiceSiteManager::serviceRequestHandlerDebug(const Request& request, Response& response) {
    ServiceSiteManager* that = ServiceSiteManager::getInstance();

    ServiceDebugResponse response_obj;
    
    ServiceSiteLibServiceResponse service_site_response;
    service_site_response.code = 0;
    service_site_response.error = "ok";

    response_obj.lib_ver = get_lib_ver();

    {
        pthread_rwlock_guard lock(that->messageSubscriberList_mutex, false);

        for (auto& item : ServiceSiteManager::getInstance()->messageSubscriberList) {
            ServiceDebugResponseMessageSubscriberListItem message_subscriber_list_item;

            message_subscriber_list_item.message_id = item.messageId;

            for (const auto& sub_item : item.getSiteMessageSubscriberSiteHandlePlist()) {
                //NOTE: 这里改为放ID(字段site_handle_list_dbg), 引用到后面的数组上, 放完整版本如果后续再增加字段的话, 调试的时候看起来太麻烦了
                //但需要注意的是订阅持久化也是用的这个类型(字段site_handle_list), 可能要拆分下
                message_subscriber_list_item.site_handle_list_dbg.push_back(sub_item->mac + "/" + sub_item->site_id);
            }

            response_obj.message_subscriber_list.push_back(message_subscriber_list_item);
        }

        for (const auto& item : ServiceSiteManager::getInstance()->messageSubscriberSiteHandlePList) {
            ServiceDebugResponseMessageSubscriberListItemSiteHandleListItem site_handle_list_item;
            site_handle_list_item.from_handle(*item);

            response_obj.message_subscriber_site_handle_list.push_back(site_handle_list_item);
        }
    }

    service_site_response.response = response_obj.to_json();

    response.set_content(service_site_response.to_json().dump(), "text/json");

    return RET_CODE_OK;
}

void ServiceSiteManager::messageHandlerRegisterAgain(const Request& request) {
    // 重新注册
    ServiceSiteManager::getInstance()->registerSite();
}

void ServiceSiteManager::http_exception_handler(const Request& request, Response& response, std::exception_ptr ep) {
    ServiceSiteManager::getInstance()->handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "http_exception_handler request.method: " + request.method);
    ServiceSiteManager::getInstance()->handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "http_exception_handler request.path: " + request.path);
    ServiceSiteManager::getInstance()->handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "http_exception_handler request.body: " + request.body);

    try {
        if (ep) {
            std::rethrow_exception(ep);
        }
    } catch(const std::exception& e) {
        ServiceSiteManager::getInstance()->handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "http_exception_handler exception: " + string(e.what()));
    }
    response.status = 500;
}

ServiceSiteManager::ServiceSiteManager() {
    if (get_uid(dev_mac) != 0) {
        ServiceSiteManager::getInstance()->handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "ServiceSiteManager::getInstance()->get_uid() != 0");
        exit(-1);
    }

    // 获取本机 ip
    get_self_ip();

    messageSubscriberConfigPath = ServiceSiteManager::MESSAGE_SUBSCRIBER_CONFIG_FILE_PATH;

    // 创建线程
	handle_message_thread = std::thread(ServiceSiteManager::handle_message_thread_func);

    // 设置异常 handler, 发生异常时打印
    server.set_exception_handler(ServiceSiteManager::http_exception_handler);

    registerServiceRequestHandler(SERVICE_ID_GET_SERVICE_LIST, ServiceSiteManager::serviceRequestHandlerGetServiceList);
    registerServiceRequestHandler(SERVICE_ID_GET_MESSAGE_LIST, ServiceSiteManager::serviceRequestHandlerGetMessageList);
    registerServiceRequestHandler(SERVICE_ID_SUBSCRIBE_MESSAGE, ServiceSiteManager::serviceRequestHandlerSubscribeMessage);
    registerServiceRequestHandler(SERVICE_ID_UNSUBSCRIBE_MESSAGE, ServiceSiteManager::serviceRequestHandlerUnsubscribeMessage);

    registerServiceRequestHandler(SERVICE_ID_DEBUG, ServiceSiteManager::serviceRequestHandlerDebug);
}

void ServiceSiteManager::handle_message_thread_func(void) {
    while (true) {
        std::unique_ptr<msg_queue_item_slim> message_pair;
        try
        {
            ServiceSiteManager::getInstance()->handle_mesage(message_pair);
        } catch(const std::exception& e) {
            ServiceSiteManager::getInstance()->handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "handle_message exception: " + string(e.what()));
            sleep(1);
        }
    }
}

void ServiceSiteManager::handle_mesage(std::unique_ptr<msg_queue_item_slim>& message_pair) {
    {
        {
            std::unique_lock<std::mutex> lock(message_queue_mutex);
            while (message_queue.empty())
            {
                message_queue_sem.wait(lock);
            }
            message_pair = std::move(message_queue.front());
            // 出队
            message_queue.pop();
        }
        if (message_pair == nullptr)
            return;
        const MessageT& message_item = message_pair->msg;
        Request request;
        request.body = std::move(message_pair->body);
        request.remote_addr = std::move(message_pair->remote_addr);

        MessageHandler handler = nullptr;
        int msg_handled = 0;
        {
            // 兼容老版本没传from的情况, 这种情况下, 检测如果注册了多个相同ID的处理, 给个WARN并跳过后续处理(仅处理第一个)
            // 最终匹配是通配符匹配的情况下, 也检查下是否还存在其他匹配, 避免出现静默的优先级错误
            bool compat_no_from = message_item.from.empty();
            string last_from;
            string last_msg;
            pthread_rwlock_guard lock(init_mutex, false);
            {
                for (const auto &x : messageHandlers) {
                    const auto &handler_message_id = x.first;

                    bool match_all_msg = ("*" == handler_message_id.messageId);
                    if (handler_message_id.messageId != message_item.messageId)
                        continue;

                    bool match_all_from = compat_no_from || ("*" == handler_message_id.from);
                    if (!match_all_from && handler_message_id.from != message_item.from)
                        continue;

                    if (msg_handled != 0)
                    {
                        msg_handled++;
                        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "Receive message with no from and mutli matches: msg=" +
                            message_item.from + "/" + message_item.messageId + "(" + to_string(msg_handled) + "), last=" +
                            last_from + "/" + last_msg + ", cur=" +
                            handler_message_id.from + "/" + handler_message_id.messageId +
                            "<-" + request.body);
                        continue;
                    }
                    last_from = handler_message_id.from;
                    last_msg = handler_message_id.messageId;
                    handler = x.second;
                    msg_handled = 1;
                    if (!match_all_from && !match_all_msg)
                        break;
                }
            }
        }
        if (handler != nullptr)
            handler(request);
        else if (msg_handled == 0)
            handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "Receive message with no handler registed: messageId=" + message_item.messageId + ", from_site_id=" + message_item.from + "<-" + request.body);
    }
}

int ServiceSiteManager::get_uid_str(string& net_dev, string& mac_str) {
	uint8_t mac[6];

	char temp_buf[80];

	int ret = get_mac(mac, net_dev.c_str());
	if (ret == 0) {
		sprintf(temp_buf, "%02X%02X%02X%02X%02X%02X", mac[0], mac[1], mac[2], mac[3], mac[4], mac[5]);
		mac_str = string(temp_buf);

		return 0;
	}

	return -1;
}

int ServiceSiteManager::get_uid(string& mac_str) {
	int ret;

	string net_dev_list[] = {
#if __x86_64__
		// 为host=linux的情况添加
		"enp0s3",
		"ens33",
#endif
		"eth0",
		"wlan0",
		"wifi0",
		"eno1"
	};

	string uid_str = "";

	for (auto& item : net_dev_list) {
		ret = get_uid_str(item, mac_str);
		if (ret == 0) {
			return 0;
		}
	}

	return -1;
}

int ServiceSiteManager::get_mac(uint8_t binMAC[6], const char* net_dev) {
    int sock;
    struct ifreq ifr;
    char* puc;
    memset(binMAC, 0, 6);
    sock = socket(AF_INET, SOCK_DGRAM, 0);
    if (sock == -1) {
        perror("socket");
        return -1;
    }
    strcpy(ifr.ifr_name, net_dev);
    if (ioctl(sock, SIOCGIFHWADDR, &ifr) < 0) {
        close(sock);
        perror("ioctl");
        return -1;
    }
    puc = ifr.ifr_hwaddr.sa_data;
    close(sock);
    memcpy(binMAC, puc, 6);
    return 0;
}

int ServiceSiteManager::popen_result(string cmd, string& result) {
	FILE *fp = NULL;
	const int BUF_SIZE = 255;
    char buf[BUF_SIZE + 1];
    int ret;

    fp = popen(cmd.c_str(), "r");

    while ((ret = fread(buf, 1, BUF_SIZE, fp)) > 0) {
		buf[ret] = 0;
        result += string(buf);
    }

    pclose(fp);

	return 0;
}

int ServiceSiteManager::get_self_ip(void) {
    string result;
    
    int ret = popen_result("ip addr show", result);

    if (ret == 0) {
        std::regex e(".*inet (.*)/.*");

        // std::cout << result << std::endl;

        std::sregex_iterator iter(result.begin(), result.end(), e);
        std::sregex_iterator end;

        while(iter != end)
        {
            if (iter->size() == 2) {
                string ip = (*iter)[1];
    
                if (ip.find("127.0.0.1") == std::string::npos) {
                    if (ip.find("169.254") != 0) {
                        // std::cout << "ip: " << ip << std::endl;
                        self_ip = ip;
                        return 0;
                    }
                }
            }

            ++iter;
        }
    }
    
    handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "get_self_ip error.");
    return -1;
}

ServiceSiteManager::msg_queue_item_slim::msg_queue_item_slim(MessageT& msg_p, const Request& req_p)
{
    msg = std::move(msg_p);
    //减少复制整个Request的成本, 但是req_p是const无法使用move
    body = req_p.body;
    remote_addr = std::move(req_p.remote_addr);
}

void ServiceSiteManager::rawHttpRequestHandler(const Request& request, Response& response) {
    ServiceSiteManager* that = ServiceSiteManager::getInstance();
    // 线程锁, 对象析构时解锁
    // std::lock_guard<std::mutex> lockGuard(that->http_request_mutex);

    // that->handle_log(LOG_LEVEL_INFO, request.body);
#if SITE_CFG_ENABLE_TRACE
    that->handle_log(ServiceSiteManager::LOG_LEVEL_INFO, __FILE__, __LINE__, "rawHttpRequestHandler: (`" + request.body + "`)");
#endif

    auto http_request_json = json::parse(request.body, /*cb*/nullptr, /*allow_exceptions*/false, /*ignore_comments*/false);
    if (!http_request_json.is_object()) {
        //解析错误为is_discarded
        response.set_content(ERROR_RESPONSE_JSON_FORMAT, "text/json");
        return;
    }

    //nlohmann在访问不存在的对象时会插入一个null, 这里我们希望能为消息(message_id)的情况保持原始语义
    if (http_request_json.contains("service_id")) {
        json &srv_id_item = http_request_json["service_id"];
        if (srv_id_item.is_string())
        {
            // Service
            string service_id = srv_id_item;
            pthread_rwlock_guard lock(that->init_mutex, false);
            for (const auto& x : that->serviceRequestHandlers) {
                const auto& handler_service_id = x.first;

                if (handler_service_id == service_id) {
                    const auto handler = x.second;
                    if (handler == nullptr)
                        break;
                    lock.unlock();

                    int code = handler(request, response);
                    if (code == 0) {
                        // 完成处理
                        return;
                    }

                    ServiceSiteLibServiceResponse service_response;
                    service_response.code = -5;
                    service_response.error = "request handler error with code = " + to_string(code);

                    // code 出错
                    response.set_content(service_response.to_json().dump(), "text/json");
                    that->handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "handler error: " + to_string(code));
                    return;
                }
            }

            // 没有匹配的 handler
            that->handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "no handler for " + request.remote_addr + " -> service_id: " + service_id);
            response.set_content(ERROR_RESPONSE_NO_REQUEST_HANDLER_MATCH, "text/json");
            return;
        }
    }

    // Message
    json &msg_id_item = http_request_json["message_id"];
    if (msg_id_item.is_string()) {
        string message_id = msg_id_item;

        MessageT message_item;
        message_item.messageId = message_id;

        if (http_request_json.contains("from")) {
            json &from_item = http_request_json["from"];
            if (from_item.is_string())
                message_item.from = from_item;
        }

        // 入队
        auto data = std::unique_ptr<msg_queue_item_slim>(new msg_queue_item_slim(message_item, request));
        {
            std::unique_lock<std::mutex> lock(that->message_queue_mutex);
            that->message_queue.push(std::move(data));
            that->message_queue_sem.notify_one();
        }

        response.set_content("{}", "text/json");
        return;
    }

    response.set_content(ERROR_RESPONSE_NOT_SERVICE_OR_MESSAGE, "text/json");
}

int ServiceSiteManager::start(void) {
    server.Post("/", ServiceSiteManager::rawHttpRequestHandler);

    loadMessageSubscriber();

    server.set_tcp_nodelay(true);
    server.set_keep_alive_max_count(100);
    // server.set_read_timeout(0, 500000);
    // server.set_write_timeout(0, 500000);

    handle_log(LOG_LEVEL_INFO, __FILE__, __LINE__, "Site `" + siteId + "` start, http listen port: " + to_string(serverPort) + ", self_ip=" + self_ip + ", mac=" + dev_mac);
    if (!server.listen("0.0.0.0", serverPort)) {
        return RET_CODE_ERROR_START_SERVER;
    }

    return RET_CODE_OK;
}

void  service_site_ping_thread(string siteId, int serverPort) {
    AccessServiceSiteQuerySitePingRequest request;

    request.site_id = siteId;
    request.port = serverPort;

    ServiceSiteLibServiceRequest service_site_requeset;

    service_site_requeset.service_id = ServiceSiteManager::ACCESS_SERVICE_ID_SITE_PING;
    service_site_requeset.to = ServiceSiteManager::QUERY_SITE_ID;
    service_site_requeset.request = request.to_json();

    json request_json = service_site_requeset.to_json();
    json response_json;

    while (true) {
        ServiceSiteManager::getInstance()->accessServiceOjm(ServiceSiteManager::QUERY_SITE_IP, ServiceSiteManager::QUERY_SITE_PORT, request_json, response_json);

        sleep(ServiceSiteManager::PING_PER_SECONDS);
    }
}

int ServiceSiteManager::startByRegister(void) {
    int ret;

    // 注册站点
    while (true) {
        ret = registerSite();
        if (ret == RET_CODE_OK) {
            break;
        }
        else {
            handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "registerSite error ret = " + to_string(ret));
            sleep(2);
        }
    }

    // 消息列表
	std::vector<string> messageIdList;
    messageIdList.push_back(ServiceSiteManager::QUERY_SITE_MESSAGE_ID_REGISTER_AGAIN);

    // 注册消息处理函数
    registerMessageHandler(ServiceSiteManager::QUERY_SITE_MESSAGE_ID_REGISTER_AGAIN, ServiceSiteManager::QUERY_SITE_ID, ServiceSiteManager::messageHandlerRegisterAgain);

    // 订阅消息
    while (true) {
        ret = subscribeMessage(ServiceSiteManager::QUERY_SITE_IP, ServiceSiteManager::QUERY_SITE_PORT, ServiceSiteManager::QUERY_SITE_ID, messageIdList);
        if (ret == RET_CODE_OK) {
            break;
        }
        else {
            handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "subscribeMessage error ret = " + to_string(ret));
            sleep(2);
        }
    }

    pingThread = std::thread(service_site_ping_thread, siteId, serverPort);

    return start();
}

int ServiceSiteManager::registerMessageHandler(string messageId, string from_site_id, MessageHandler handler) {

    //NOTE: 注意通配符的匹配优先级不是固定最后, 而是按初次添加的先后顺序进行的匹配
    if (from_site_id.empty())
    {
        //警告并回退到通配符
        from_site_id = "*";
        handle_log(LOG_LEVEL_WARN, __FILE__, __LINE__, "registerMessageHandler, from_site_id is empty, message from all sites will be matched: messageId=" + messageId);
    }
    if (messageId.empty())
    {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "registerMessageHandler, messageId is empty: from_site_id=" + from_site_id);
    }

    pthread_rwlock_guard lock(init_mutex, true);
    for (auto &x : messageHandlers) {
        const auto &handler_message_id = x.first;

        if (handler_message_id.messageId != messageId)
            continue;
        if (handler_message_id.from != from_site_id)
            continue;

        x.second = handler;
        handle_log(LOG_LEVEL_WARN, __FILE__, __LINE__, "registerMessageHandler, dup message handler registed, only the last one will be used: messageId=" + messageId + ", from_site_id=" + from_site_id);
        return RET_CODE_OK;
    }

    MessageT messageId_t;
    messageId_t.from = from_site_id;
    messageId_t.messageId = messageId;
    messageHandlers.push_back(std::make_pair(messageId_t, handler));

    return RET_CODE_OK;
}

static bool _msg_key_chk(ServiceSiteManager* that, json& obj, const string& key, const string val, const string message, int msg_chk_mode)
{
    //如果内部值也无效, 那么什么也做不了, 考虑直接跳过检查
    if (val.empty())
        return false;
    bool field_rewrite = true;
    bool field_warn = false;

    if (!obj.contains(key))
    {
        // field_rewrite = true;
    }
    else
    {
        json &msg_id_item = obj[key];
        if (msg_id_item.is_null())
        {
            // field_rewrite = true;
        }
        else if (msg_id_item.is_string())
        {
            const string tmp = (string)msg_id_item;
            if (tmp.empty())
            {
                // field_rewrite = true;
            }
            else if (tmp != val)
            {
                if (msg_chk_mode == 1)
                {
                    field_rewrite = true;
                }
                else
                {
                    field_warn = true;
                    field_rewrite = true;
                }
            }
            else
            {
                field_rewrite = false;
            }
        }
        else
        {
            if (msg_chk_mode == 1)
            {
                field_rewrite = true;
            }
            else
            {
                field_warn = true;
                field_rewrite = true;
            }
        }
    }
    if (field_warn)
    {
        that->handle_log(ServiceSiteManager::LOG_LEVEL_WARN, __FILE__, __LINE__, "publishMsg: `" + key + "` mismatch, " + val + "<-" + message);
    }
    if (field_rewrite)
    {
        obj[key] = val;
        return true;
    }
    return false;
}

void ServiceSiteManager::publishMessage(string messageId, string message, bool is_high_priority, int msg_chk_mode) {
#if SITE_CFG_ENABLE_TRACE
    this->handle_log(ServiceSiteManager::LOG_LEVEL_INFO, __FILE__, __LINE__, "publishMessage: (`" + messageId + "` , `" + message + "`)");
#endif
    if (msg_chk_mode != 2)
    {
        auto obj = json::parse(message, /*cb*/nullptr, /*allow_exceptions*/false, /*ignore_comments*/false);
        if (!obj.is_object())
        {
            handle_log(LOG_LEVEL_WARN, __FILE__, __LINE__, "publishMsg: not json obj<-" + message);
        }
        else
        {
            // 如果外部未设置, 插入下必须的字段(外部如果设置了任何值, 当前版本暂不验证其正确性)
            bool mod = false;

            mod |= _msg_key_chk(this, obj, "message_id", messageId, message, msg_chk_mode);
            mod |= _msg_key_chk(this, obj, "from", siteId, message, msg_chk_mode);

            if (mod)
                message = obj.dump();
            if (messageId.empty() && obj.contains("message_id") && obj["message_id"].is_string())
            {
                messageId = (string)obj["message_id"];
            }
        }
    }
    {
        //当尝试发布一个未注册(不是未订阅)的消息时, 打印警告
        bool msg_is_registed = false;
        pthread_rwlock_guard lock(init_mutex, false);
        for (const auto& item : messageIds) {
            if (messageId != item.messageId)
                continue;
            
            msg_is_registed = true;
            break;
        }
        if (!msg_is_registed)
        {
            this->handle_log(ServiceSiteManager::LOG_LEVEL_WARN, __FILE__, __LINE__, "Try to publish a message which is NOT registed");
        }
    }

    pthread_rwlock_guard lock(messageSubscriberList_mutex, false);
    for (auto& messageSubscriberItem : messageSubscriberList) {
        if (messageId == messageSubscriberItem.messageId) {
            for (auto& siteHandlePItem : messageSubscriberItem.getSiteMessageSubscriberSiteHandlePlist()) {
                siteHandlePItem->sendMessage(message, is_high_priority);
            }
            break;
        }
    }
}

int ServiceSiteManager::subscribeMessage(string ip, int port, string to_site_id, std::vector<string>& messageIdList) {
    AccessServiceAnySubscribeMessageRequest request;

    for (auto& messageId : messageIdList) {
        bool found = false;
        for (auto& item2 : messageHandlers){
            const auto &handler_message_id = item2.first;
            if (handler_message_id.messageId != messageId)
                continue;

            if (handler_message_id.from != to_site_id)
                continue;

            found = true;
            break;
        }
        if (!found)
        {
            handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "subscribeMessage: try to subscribe an message without handler registed<-" + ServiceSiteManager::dbg_tag(ip, port, to_site_id, false) + "/" + messageId);
        }
    }

    //复制一份, 不要改变外部传入的列表
    request.message_list = messageIdList;
    // 自身 ip
    if (ip == "127.0.0.1") {
        request.ip = "127.0.0.1";
    }
    else {
        // 获取 ip
        if (0 != get_self_ip()) {
            handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "subscribeMessage: get_self_ip() error.");
            return -300;
        }

        //TODO: 旧版中有部分站点屏蔽了来自127.0.0.1的事件, 看后续是否需要回退这个
        // 如果对端IP和本机局域网IP一致, 将其重写为127.0.0.1以避免后续DHCP等原因导致的IP改变
        // 主从和配置站点都不会同步我们持久化的IP数据, 所以将其映射回本机应该是安全的
        if (ip == self_ip) {
            handle_log(LOG_LEVEL_INFO, __FILE__, __LINE__, "subscribeMessage: remap local IP to loopback : " + ip + "-> 127.0.0.1");
            ip = "127.0.0.1";
            request.ip = "127.0.0.1";
        }
        else
        {
            request.ip = self_ip;
        }
    }
    request.port = serverPort;
    request.site_id = siteId;
    request.mac = dev_mac;

    ServiceSiteLibServiceRequest service_site_requeset;

    service_site_requeset.service_id = ACCESS_SERVICE_ID_SUBSCRIBE_MESSAGE;
    service_site_requeset.to = to_site_id;
    service_site_requeset.request = request.to_json();

    json request_json = service_site_requeset.to_json();
    json response_json;

    int ret = accessServiceOjm(ip, port, request_json, response_json);
    if (ret != RET_CODE_OK) {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "accessService error, code = " + to_string(ret));
        return -301;
    }

    AccessServiceAnySubscribeMessageResponse response;

    ret = response.from_json(response_json);
    if (ret != 0) {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "response from_json error, code = " + to_string(ret));
        return -302;
    }

    return RET_CODE_OK;
}

int ServiceSiteManager::unsubscribeMessage(string ip, int port, string to_site_id, std::vector<string> messageIdList) {
    AccessServiceAnyUnsubscribeMessageRequest request;

    //复制一份, 不要改变外部传入的列表
    request.message_list = messageIdList;
    request.port = serverPort;
    request.site_id = siteId;
    request.mac = dev_mac;

    ServiceSiteLibServiceRequest service_site_requeset;

    service_site_requeset.service_id = ACCESS_SERVICE_ID_UNSUBSCRIBE_MESSAGE;
    service_site_requeset.to = to_site_id;
    service_site_requeset.request = request.to_json();

    json request_json = service_site_requeset.to_json();
    json response_json;

    int ret = accessServiceOjm(ip, port, request_json, response_json);
    if (ret != RET_CODE_OK) {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "unsubscribeMessage: accessService error, code = " + to_string(ret));
        return -301;
    }

    AccessServiceAnyUnsubscribeMessageResponse response;

    ret = response.from_json(response_json);
    if (ret != 0) {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "unsubscribeMessage: response from_json error, code = " + to_string(ret));
        return -302;
    }

    return RET_CODE_OK;
}

int ServiceSiteManager::createDir(string sPathName) {
    if (access(sPathName.c_str(), F_OK) == 0)
    {
        return 0;
    }

    int res = 0;
    char* DirName = strdup(sPathName.c_str());
    int len = strlen(DirName);
    for (int i = 1; i < len; i++) {
        if (DirName[i] == '/') {
            DirName[i] = 0;
            if (access(DirName, F_OK) != 0) {
                if (mkdir(DirName, 0755) == -1) {
                    handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "mkdir error");

                    res = -1;
                    goto clean_up;
                }
            }
            DirName[i] = '/';
        }
    }

clean_up:
    free(DirName);
    return res;
}

void ServiceSiteManager::handle_log(int level, string file, int line, string message) {
    size_t pos = file.find_last_of('/');
    if (pos != string::npos)
        file = file.substr(pos + 1);
    string log_message = "[" + file + ":" + to_string(line) + "]" + message;

    if (log_handler != nullptr) {
        log_handler(level, log_message);
        return;
    }

    switch (level) {
    case LOG_LEVEL_INFO:
        printf("\033[0;32m[INFO]\033[0m %s\n", log_message.c_str());
        break;
    case LOG_LEVEL_WARN:
        printf("\033[1;33m[WARN]\033[0m %s\n", log_message.c_str());
        break;
    case LOG_LEVEL_ERROR:
        printf("\033[0;31m[ERROR]\033[0m %s\n", log_message.c_str());
        break;
    default:
        break;
    }
}

void ServiceSiteManager::registerLogHandler(LogHandler handler) {
    log_handler = handler;
}

int ServiceSiteManager::accessServiceOjm(string ip, int port, json& request, json& response) {
    json json_response;
    int ret = accessService(ip, port, request, json_response);
    if (ret != RET_CODE_OK) {
        return ret;
    }

    ServiceSiteLibServiceResponse service_site_lib_service_response;

    ret = service_site_lib_service_response.from_json(json_response);
    if (ret != 0) {
        string to_site_id = "***UNKNOWN***";
        if (request.contains("to") && request["to"].is_string())
        {
            to_site_id = request["to"];
        }
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, ServiceSiteManager::dbg_tag(ip, port, to_site_id, true) + "response from_json error, code = " + to_string(ret)
#if SITE_CFG_LOG_WITH_DATA
            + ", req=" + request.dump() + ", resp=" + json_response.dump()
#endif
            );
        return -201;
    }

    if (service_site_lib_service_response.code != 0) {
        string to_site_id = "***UNKNOWN***";
        if (request.contains("to") && request["to"].is_string())
        {
            to_site_id = request["to"];
        }
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, ServiceSiteManager::dbg_tag(ip, port, to_site_id, true) + "response error, code = " + to_string(service_site_lib_service_response.code) 
            + " error = " + service_site_lib_service_response.error
#if SITE_CFG_LOG_WITH_DATA
            + ", req=" + request.dump() + ", resp=" + json_response.dump()
#endif
            );
        return -202;
    }

    response = service_site_lib_service_response.response;

    return 0;
}

int ServiceSiteManager::accessService(string ip, int port, json& json_request, json& json_response) {
    Client cli(ip, port);
    cli.set_connection_timeout(3, 0);

    auto res = cli.Post("/", json_request.dump(), "text/json");
    if (!res) {
        string to_site_id = "***UNKNOWN***";
        if (json_request.contains("to") && json_request["to"].is_string())
        {
            to_site_id = json_request["to"];
        }
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, ServiceSiteManager::dbg_tag(ip, port, to_site_id, true) + "client connect error.");
        return -101;
    }

    if (res->status != 200) {
        string to_site_id = "***UNKNOWN***";
        if (json_request.contains("to") && json_request["to"].is_string())
        {
            to_site_id = json_request["to"];
        }
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, ServiceSiteManager::dbg_tag(ip, port, to_site_id, true) + "http status = " + to_string(res->status) + ", error."
#if SITE_CFG_LOG_WITH_DATA
            + ", req=" + json_request.dump()
#endif
            );
        return -102;
    }

    if (!json::accept(res->body)) {
        string to_site_id = "***UNKNOWN***";
        if (json_request.contains("to") && json_request["to"].is_string())
        {
            to_site_id = json_request["to"];
        }
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, ServiceSiteManager::dbg_tag(ip, port, to_site_id, true) + "json_response json format error"
#if SITE_CFG_LOG_WITH_DATA
            + ", req=" + json_request.dump() + ", resp=" + res->body
#endif
            );
        return -103;
    }

    json_response = json::parse(res->body);

    return RET_CODE_OK;
}

int ServiceSiteManager::getMessageList(string ip, int port, string to_site_id, std::vector<string>& messageIdList) {
    AccessServiceAnyGetMessageListRequest request;

    ServiceSiteLibServiceRequest service_site_requeset;

    service_site_requeset.service_id = ACCESS_SERVICE_ID_GET_MESSAGE_LIST;
    service_site_requeset.to = to_site_id;
    service_site_requeset.request = request.to_json();

    json request_json = service_site_requeset.to_json();
    json response_json;

    int ret = accessServiceOjm(ip, port, request_json, response_json);
    if (ret != RET_CODE_OK) {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "accessService error, code = " + to_string(ret));
        return -301;
    }

    AccessServiceAnyGetMessageListResponse response;

    ret = response.from_json(response_json);
    if (ret != 0) {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "response from_json error, code = " + to_string(ret));
        return -302;
    }

    for (auto& item : response.message_list) {
        messageIdList.push_back(item.message_id);
    }

    return RET_CODE_OK;
}

MessageSubscriber::MessageSubscriber(string pMessageId) {
    messageId = pMessageId;
}

bool MessageSubscriber::addSiteMessageSubscriberSiteHandleP(MessageSubscriberSiteHandle* pMessageSubscriberSiteHandle) {
    bool need_save = false;

    MessageSubscriberSiteHandle* temp = NULL;
    for (auto& item : siteMessageSubscriberSiteHandlePlist) {
        if (item == pMessageSubscriberSiteHandle) {
            // 已存在
            temp = item;
            break;
        }
    }

    if (temp == NULL) {
        // 不存在 添加
        siteMessageSubscriberSiteHandlePlist.push_back(pMessageSubscriberSiteHandle);

        need_save = true;
    }

    return need_save;
}

void MessageSubscriber::delSiteMessageSubscriberSiteHandleP(MessageSubscriberSiteHandle* pMessageSubscriberSiteHandle) {
    auto iter = siteMessageSubscriberSiteHandlePlist.begin();
    while (iter != siteMessageSubscriberSiteHandlePlist.end()) {
        if (pMessageSubscriberSiteHandle == *iter) {
            iter = siteMessageSubscriberSiteHandlePlist.erase(iter);
        } else {
            ++iter;
        }
    }
}

std::vector<MessageSubscriberSiteHandle*> MessageSubscriber::getSiteMessageSubscriberSiteHandlePlist(void) {
    return siteMessageSubscriberSiteHandlePlist;
}

void message_subscriber_site_handle_send_message_thread(MessageSubscriberSiteHandle& messageSubscriberSiteHandle) {
    while (true) {
        messageSubscriberSiteHandle.sendFromThread();
    }
}

MessageSubscriberSiteHandle::MessageSubscriberSiteHandle(string p_mac, string p_site_id, int p_port, string p_ip) {

static_assert(MessageSubscriberSiteHandle::SEND_RETRY_HARD_LIMIT >= MessageSubscriberSiteHandle::MAX_SEND_RETRY, "INVAILD CONFIG");
static_assert(MessageSubscriberSiteHandle::MAX_SEND_RETRY > 0, "INVAILD CONFIG");
static_assert(MessageSubscriberSiteHandle::MAX_QUEUE_SIZE > 0, "INVAILD CONFIG");
static_assert(MessageSubscriberSiteHandle::MAX_HIGH_PRIORITY_QUEUE_SIZE > 0, "INVAILD CONFIG");
static_assert(SITE_CFG_SEND_CNT >= 0, "INVAILD CONFIG");

    mac = p_mac;
    site_id = p_site_id;
    port = p_port;
    ip = p_ip;

    require_fresh = 1;

    // 创建线程
	sendMessageThreadP = new std::thread(message_subscriber_site_handle_send_message_thread, std::ref(*this));
}

void MessageSubscriberSiteHandle::sendFromThread(void) {
    string message;

    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        do
        {
            if (!high_priority_queue.empty())
            {
                // 优先发送高优先级
                message = std::move(high_priority_queue.front());
                high_priority_queue.pop();
            }
            else if (!queue.empty())
            {
                message = std::move(queue.front());
                queue.pop();
            }
            else
            {
                queue_sem.wait(lock);
                continue;
            }
            break;
        } while (true);
    }

    if (message.empty()) {
        return;
    }
    if (retry_count == SEND_RETRY_SUSPEND)
    {
        //可能是之前队列里残留的, 清理下
        stats_cnt_drop[STAT_DROP_IDX_SUSPEND] ++;
        do_suspend();
        return;
    }

    int err = STAT_DROP_IDX_CONN;
    for (int resend = 0; ; resend++)
    {
        // 每次重连时再检查下cli信息有无更新
        if (require_fresh == 1 || cli == nullptr) {
            //NOTE: cli会自己处理重连, 因此连接断开时无需特殊处理
            string old_addr = "";
            if (cli != nullptr) {
                old_addr = cli->host() + ":" + to_string(cli->port());
                cli->stop();
                delete cli;
                cli = nullptr;
            }
            {
                std::unique_lock<std::mutex> lock(queue_mutex);

                cli = new Client(ip, port);
                cli->set_keep_alive(true);
                cli->set_tcp_nodelay(true);
                cli->set_connection_timeout(1, 0);
                // cli->set_write_timeout(1, 0);
                // cli->set_read_timeout(1, 0);

                retry_count = 0;
                stats_cnt_send = 0;
                stats_cnt_drop.fill(0);

                require_fresh = 0;
                //如果在重发过程中产生了重连, 重置下计数(这应该是小概率事件, 忽略会无限次发生的情况)
                resend = 0;
            }
            if (old_addr.empty())
                ServiceSiteManager::getInstance()->handle_log(ServiceSiteManager::LOG_LEVEL_INFO, __FILE__, __LINE__, dbg_tag() + "Connect to server");
            else
                ServiceSiteManager::getInstance()->handle_log(ServiceSiteManager::LOG_LEVEL_WARN, __FILE__, __LINE__, dbg_tag() + "Switch to new server from " + old_addr);
        }
        if (resend != 0 && retry_count >= MAX_SEND_RETRY)
        {
            // 为长期离线情况, 提供一条快速失败路径(跳过重试)
            break;
        }
        if (resend > SITE_CFG_SEND_CNT)
            break;

        auto res = cli->Post("/", message, "text/json");
        if (!res) {
            // TODO: 如果不是连接超时的话, 等待100~200ms
            // TODO: 现在syn包的重传时间好像是1s, 可能太慢和set_connection_timeout的配合不太好
            // NOTE: 只要重发期间出现过一次http错误, 那么最终错误码就按http错误计算
            // err = STAT_DROP_IDX_CONN;
            ServiceSiteManager::getInstance()->handle_log(ServiceSiteManager::LOG_LEVEL_ERROR, __FILE__, __LINE__, dbg_tag()
                + "retry=" + to_string(retry_count) + ", resend=" + to_string(resend) 
                + ", client connect error. ");
        }
        else {
            if (res->status == 200) {
                // 成功, 将累计失败计数器清零, 并退出
                retry_count = 0;
                stats_cnt_send ++;
                return;
            }
            err = STAT_DROP_IDX_HTTP;
            // TODO: http错误是否应该触发长期失败和SUSPEND??
            // TODO: 我们真的打算重试http错误么? 这并不是幂等操作
            ServiceSiteManager::getInstance()->handle_log(ServiceSiteManager::LOG_LEVEL_ERROR, __FILE__, __LINE__, dbg_tag()
                + "retry=" + to_string(retry_count) + ", resend=" + to_string(resend)
                + ", http error, status = " + to_string(res->status)
#if SITE_CFG_LOG_WITH_DATA
                + ", req=" + message
#endif
                );
        }
    }

    stats_cnt_drop[err]++;
    if (retry_count < 0)
        return;

    if (retry_count == 0 || failed_ts <= TS_YEAR_2020)
    {
        failed_ts = get_ts();
    }
    if (retry_count < SEND_RETRY_HARD_LIMIT)
        retry_count ++;
    if (need_suspend())
    {
        do_suspend();
        return;
    }
    if (retry_count < MAX_SEND_RETRY)
    {
        return;
    }

    {
        uint32_t drop = empty_queue();
        if (drop != 0)
            stats_cnt_drop[STAT_DROP_IDX_FAST] += drop;

        // NOTE: 将日志操作移出到锁外部
        ServiceSiteManager::getInstance()->handle_log(ServiceSiteManager::LOG_LEVEL_ERROR, __FILE__, __LINE__, 
                dbg_tag() + 
                "sendRetryCount = " + to_string(retry_count) + ", error, drop all " + to_string(drop + 1) + " messages.");
    }
}

uint32_t MessageSubscriberSiteHandle::empty_queue(){
    uint32_t drop = 0;
    std::unique_lock<std::mutex> lock(queue_mutex);

    if (!high_priority_queue.empty())
    {
        drop += high_priority_queue.size();
        std::queue<string> empty;
        high_priority_queue.swap(empty);
    }
    if (!queue.empty())
    {
        drop += queue.size();
        std::queue<string> empty;
        queue.swap(empty);
    }
    return drop;
}

string MessageSubscriberSiteHandle::dbg_tag(){
    return "[" + mac + ", " + ServiceSiteManager::dbg_tag(ip, port, site_id, false) + "]";
}

void MessageSubscriberSiteHandle::sendMessage(const string& message, bool is_high_priority) {
    int drop = 0;
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        if (retry_count == SEND_RETRY_SUSPEND)
        {
            //NOTE: 为避免无意义的日志输出, 此处不打印日志, 具体条目可通过接口查看
            //NOTE: 考虑下每小时输出一轮调试信息
            stats_cnt_drop[STAT_DROP_IDX_SUSPEND] ++;
            return;
        }

        // TODO: 现在修改为按需drop, 如果出现过于频繁导致性能不好的话考虑下设置drop的上下界
        if (is_high_priority) {
            // 高优先级
            while(high_priority_queue.size() >= MAX_HIGH_PRIORITY_QUEUE_SIZE) {
                drop++;
                high_priority_queue.pop();
            }
            high_priority_queue.push(message);
        }
        else {
            while (queue.size() >= MAX_QUEUE_SIZE) {
                drop++;
                queue.pop();
            }
            queue.push(message);
        }
        queue_sem.notify_one();
        //NOTE: 在队列溢出的情况下, 我们选择丢弃最老的条目
        stats_cnt_drop[STAT_DROP_IDX_OVERFLOW] += drop;
    }
    // NOTE: 将日志操作移出到锁外部
    if (drop > 1)
        ServiceSiteManager::getInstance()->handle_log(ServiceSiteManager::LOG_LEVEL_ERROR, __FILE__, __LINE__, dbg_tag() + (string)(is_high_priority ? "high_priority_queue" : "queue") + " is overflow, drop " + to_string(drop) + " old messages.");
}

void MessageSubscriberSiteHandle::do_suspend() {
    int pre_retry_count = retry_count;
    if (pre_retry_count != SEND_RETRY_SUSPEND)
    {
        retry_count = SEND_RETRY_SUSPEND;
        if (cli != nullptr) {
            cli->stop();
        }

        uint32_t cur_ts = get_ts();
        ServiceSiteManager::getInstance()->handle_log(ServiceSiteManager::LOG_LEVEL_ERROR, __FILE__, __LINE__, dbg_tag() 
            + "Still failed after " + to_string(pre_retry_count) + " retries and " + to_string(cur_ts - failed_ts) + " seconds, suspend until next subscribe"
            + ", last_ts=" + to_string(failed_ts) + ", cur_ts=" + to_string(cur_ts));
    }
    uint32_t drop = empty_queue();
    stats_cnt_drop[STAT_DROP_IDX_SUSPEND] += drop;
}

bool MessageSubscriberSiteHandle::refresh(string p_ip, int p_port) {
    if (port == p_port && ip == p_ip && retry_count != SEND_RETRY_SUSPEND)
        return false;
    {
        std::unique_lock<std::mutex> lock(queue_mutex);

        // 连接目标发生改变, 强制下一次重置cli
        port = p_port;
        ip = p_ip;
        require_fresh = 1;
        //强制退出休眠状态
        if (retry_count == SEND_RETRY_SUSPEND)
            retry_count = 0;
        return true;
    }
}

uint32_t MessageSubscriberSiteHandle::get_ts()
{
    return (uint32_t)(time(NULL) & 0xFFFFFFFFuL);
}

bool MessageSubscriberSiteHandle::need_suspend()
{
    if (retry_count == SEND_RETRY_SUSPEND)
        return true;
#if SITE_CFG_SUSPEND_TIME > 0
    if (retry_count >= MAX_SEND_RETRY && failed_ts >= TS_YEAR_2020)
    {
        //考虑下刚开机还没同步到时钟的情况
        //NOTE: 如果时钟未同步, 即使已经在1970年走了足够长的时间, 也不计休眠
        uint32_t cur_ts = get_ts();
        if (cur_ts >= TS_YEAR_2020 && ((int32_t)(cur_ts - failed_ts)) >= static_cast<int32_t>(SITE_CFG_SUSPEND_TIME))
        {
            return true;
        }
    }
#endif
    return false;
}

int servicesite::ServiceSiteManager::file_len(const char *filename) {
	int len = -1;
	struct stat fileStat;

	if (0 == stat(filename, &fileStat)) {
		len = fileStat.st_size;
	}

	return len;
}

void servicesite::ServiceSiteManager::saveMessageSubscriber(void) {
#if SITE_CFG_SKIP_SUB_SAVE
    return;
#endif
    json message_subscriber_list = json::array();

    //NOTE: 这里本身需要的是读锁, 但磁盘本身也需要锁, 加一个覆盖整个函数的写锁(在日志之前退出)作为互斥
    pthread_rwlock_guard lock(messageSubscriberList_mutex, true);
    for (auto& item : messageSubscriberList) {
        ServiceDebugResponseMessageSubscriberListItem message_subscriber_list_item;

        message_subscriber_list_item.message_id = item.messageId;

        for (const auto& sub_item : item.getSiteMessageSubscriberSiteHandlePlist()) {
            ServiceDebugResponseMessageSubscriberListItemSiteHandleListItem site_handle_list_item;
            site_handle_list_item.from_handle(*sub_item);

            message_subscriber_list_item.site_handle_list.push_back(site_handle_list_item);
        }

        message_subscriber_list.push_back(message_subscriber_list_item.to_json());
    }

    string config_filename = messageSubscriberConfigPath + siteId + MESSAGE_SUBSCRIBER_CONFIG_FILE;

    // handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "----%s\n", config_filename.c_str());
    // handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "----%s\n", message_subscriber_list.dump(4).c_str());

    if (0 != createDir(ServiceSiteManager::messageSubscriberConfigPath)) {
        lock.unlock();
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "createDir error: " + messageSubscriberConfigPath);
        return;
    }

    int fd = open(config_filename.c_str(), O_CREAT | O_WRONLY | O_TRUNC, 0777);
    if(fd == -1)
    {
        lock.unlock();
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "open error: " + config_filename);
        return;
    } 

    string json_data = message_subscriber_list.dump(4);
    write(fd, json_data.c_str(), json_data.size());
    fsync(fd);
    close(fd);

    lock.unlock();
    handle_log(LOG_LEVEL_INFO, __FILE__, __LINE__, "saveMessageSubscriber ok.");
}

void servicesite::ServiceSiteManager::loadMessageSubscriber(void) {
#if SITE_CFG_SKIP_SUB_SAVE
    return;
#endif
    string config_filename = messageSubscriberConfigPath + siteId + MESSAGE_SUBSCRIBER_CONFIG_FILE;

    // 检查文件是否存在
	if (access(config_filename.c_str(), 0) != 0) {
		handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "access error: " + config_filename);
		return;
	}

	// 获取文件大小
	int buf_size = file_len(config_filename.c_str());
	if (buf_size <= 0) {
		handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "file_len error: " + config_filename);
		return;
	}
	
	char* buf = new char[buf_size + 1];
	int read_len = 0;

    json message_subscriber_list;
    string message_id;
    string ip;
    int port;

	int fd = open(config_filename.c_str(), O_RDONLY);
	if(fd == -1)
	{
		handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "open error: " + config_filename);
		goto end;
	}

	read_len = read(fd, buf, buf_size + 1);

	if (read_len > buf_size || read_len < 0) {
		handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "read_len error: " + to_string(read_len));
		goto end;
	}

	buf[read_len] = '\0';

	if (!json::accept(string(buf))) {
		handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "json::accept error: " + string(buf));
		goto end;
	}

	message_subscriber_list = json::parse(buf);

    if (!message_subscriber_list.is_array()) {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "json::parse error: " + message_subscriber_list.dump());
        goto end;
    }

    // handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "----%s\n", message_subscriber_list.dump(4).c_str());
    for (json& item : message_subscriber_list) {
        ServiceDebugResponseMessageSubscriberListItem message_subscriber_list_item;

        if (message_subscriber_list_item.from_json(item) != 0) {
            handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "json::parse messageId error: " + item.dump());
            return;
        }

        for (auto& sub_item : message_subscriber_list_item.site_handle_list) {
            //这个在保存/读取或者是运行的时候跳过?
            if (sub_item.retry_count == MessageSubscriberSiteHandle::SEND_RETRY_SUSPEND)
                continue;
            handleSubscribeMessage(message_subscriber_list_item.message_id, sub_item.mac, sub_item.site_id, sub_item.port, sub_item.ip);
        }
    }

end:
	delete[] buf;
	close(fd);
	return;
}

int servicesite::ServiceSiteManager::registerSite(void) {
    AccessServiceSiteQuerySiteRegisterRequest request;

    request.site_id = siteId;
    request.summary = summary;
    request.port = serverPort;

    ServiceSiteLibServiceRequest service_site_requeset;

    service_site_requeset.service_id = ACCESS_SERVICE_ID_SITE_REGISTER;
    service_site_requeset.to = QUERY_SITE_ID;
    service_site_requeset.request = request.to_json();

    json request_json = service_site_requeset.to_json();
    json response_json;

    int ret = accessServiceOjm(ServiceSiteManager::QUERY_SITE_IP, ServiceSiteManager::QUERY_SITE_PORT, request_json, response_json);
    if (ret != RET_CODE_OK) {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "accessService error, code = " + to_string(ret));
        return -301;
    }

    AccessServiceSiteQuerySiteRegisterResponse response;

    ret = response.from_json(response_json);
    if (ret != 0) {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "response from_json error, code = " + to_string(ret));
        return -302;
    }

    return RET_CODE_OK;
}

//TODO: 处理下同一IP/端口存在多个站点的情况(如来自ESP32的http代理?, 或旧版消息不存在site_id的消息订阅数据)
bool servicesite::ServiceSiteManager::handleSubscribeMessage(string message_id, string mac, string site_id, int port, string ip) {
    bool need_save = false;
    bool is_msg_id_ok = false;

    // 检查是否支持此 消息ID
    for (auto& item : messageIds) {
        if (item.messageId == message_id) {
            is_msg_id_ok = true;
            break;
        }
    }
    // 不支持消息ID，返回false
    if (!is_msg_id_ok) {
        handle_log(LOG_LEVEL_ERROR, __FILE__, __LINE__, "handleSubscribeMessage for non-registed message: message_id=" + message_id + "<-" + mac + "," + ServiceSiteManager::dbg_tag(ip, port, site_id, false));
        return false;
    }

    MessageSubscriberSiteHandle* temp_messageSubscriberSiteHandle = NULL;
    pthread_rwlock_guard lock(messageSubscriberList_mutex, true);
    // 查找站点是否订阅过
    for (auto& item : messageSubscriberSiteHandlePList) {
        if (item->mac == mac) {
            if (item->site_id == site_id) {
                temp_messageSubscriberSiteHandle = item;
                break;
            }
        }
    }

    // 未订阅创建新的站点handle
    if (temp_messageSubscriberSiteHandle == NULL) {
        temp_messageSubscriberSiteHandle = new MessageSubscriberSiteHandle(mac, site_id, port, ip);
        messageSubscriberSiteHandlePList.push_back(temp_messageSubscriberSiteHandle);
        need_save = true;
    }
    else {
        // 更新已存在的信息
        if (temp_messageSubscriberSiteHandle->refresh(ip, port)) {
            need_save = true;
        }
    }
    
    // 检查此消息ID的订阅者是否存在
    MessageSubscriber* temp_messageSubscriber = NULL;
    for (auto& item : messageSubscriberList) {
        if (message_id == item.messageId) {
            temp_messageSubscriber = &item;
            break;
        }
    }

    // 不存在，创建订阅者
    if (temp_messageSubscriber == NULL) {
        temp_messageSubscriber = new MessageSubscriber(message_id);
        temp_messageSubscriber->addSiteMessageSubscriberSiteHandleP(temp_messageSubscriberSiteHandle);

        messageSubscriberList.push_back(*temp_messageSubscriber);
        need_save = true;
    }
    else {
        // 已存在
        if (temp_messageSubscriber->addSiteMessageSubscriberSiteHandleP(temp_messageSubscriberSiteHandle)) {
            need_save = true;
        }
    }

    return need_save;
}

string ServiceSiteManager::dbg_tag(const string& ip, int port, const string& site_id, bool with_bracket)
{
    string str = ip + ":" + to_string(port) + "/" + site_id;
    if (with_bracket)
        str = "[" + str + "]";
    return str;
}

string ServiceSiteManager::get_lib_ver()  {
    return SERVICE_SITE_LIB_VER;
}